#!/bin/bash
#设置任务异常退出
set -eo pipefail
dt='{{ execution_date | cst_ds }}'
name='T-1出港转运及时率-明细兔优达'
# 在退出时清理文件
trap "rm -f /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv && echo 'removed tmp files'" EXIT
# 创建txt文件
touch /tmp/${name}_${dt}.txt
#将中文字段名输入文件顶部
#将查询结果追加到文件中
{
  echo -e "运单号\t始发地发件中心\t始发地发件中心\t订单来源code\t订单来源\t发件代理区code\t发件代理区\t中心类型1始发2中转\t中心发件下一站code\t中心发件下一站\t发件扫描时间\t发车任务号\t规划发车时间\t实际发车时间\t下一站规划到达时间\t所属包号\t物品类型\t物品类型\t到件任务号\t到件时间\t全链路到件时间\t到件任务对应上一站机构code\t到件任务对应上一站机构\t中转时长\t应发件时间\t全链路应发件时间\t是否及时1及时2路由缺失3不及时\t不及时原因:1.路由缺失2.错发3.任务单问题4.发车晚点5.操作延误6.经停线问题7.及时\t全链路是否及时1及时2路由缺失3不及时\t全链路不及时原因:1.路由缺失2.错发3.任务单问题4.发车晚点5.操作延误6.经停线问题7.及时\t路由缺失\t错发\t目的网点\t目的网点\t路由到达下一站时间\t全链路路由到达下一站时间\t路由本站发出车最晚到车时间\t全链路路由本站发出车辆最晚到车时间\t是否经停点1是2否\t封发车时间\t管理大区编码\t管理大区名称\t业务属性 1:正班 2:加班\t中心发件下一站兜底路由中发车班次最晚到车时间\t全链路中心发件下一站兜底路由中发车班次最晚到车时间\t到件任务实际到车时间\t到件任务规划到车时间\t衔接时长\t签收代理区code\t签收代理区名称\t目的中心code\t目的中心名称\t转运中心到件扫描时间\t三段码\t任务单实际到达本站时间\t">/tmp/${name}_${dt}.txt;
  spark-sql --driver-memory 8G --num-executors 10 --executor-memory 12g --executor-cores 4  --conf spark.executor.memoryOverhead=4g --conf spark.driver.maxResultSize=3g -S -e "
with departure_transfer as (
    select billcode
         , start_code
         , start_name
         , order_source_code
         , order_source_name
         , start_agent_code
         , start_agent_name
         , center_type
         , nextstationcode
         , nextstation
         , send_scantime
         , send_shipment_no
         , planned_departure_time
         , actual_departure_time
         , next_planned_arrival_time
         , packagecode
         , goods_type_code
         , goods_type_name
         , arrival_shipment_no
         , arrival_time
         , all_line_arrival_time
         , last_network_code
         , last_network_name
         , transfer_hours
         , route_send_time
         , all_line_route_send_time
         , is_timely
         , untimely_cause
         , all_line_is_timely
         , all_line_untimely_cause
         , route_lack
         , route_error
         , end_code
         , end_name
         , r_nxt_arrival_time
         , all_line_r_nxt_arrival_time
         , r_lastes_arrival_time
         , all_line_r_lastes_arrival_time
         , stop_type
         , pack_send_time
         , mage_region_code
         , mage_region_name
         , business_attribute
         , r_next_lastes_arrival_time
         , all_r_next_lastes_arrival_time
         , arrival_actual_arrival_time
         , arrival_planned_arrival_time
         , reserved_seconds
         , sign_agent_code
         , sign_agent_name
         , end_center_code
         , end_center_name
         , center_arrvial_time
         , replace(terminal_dispatch_code,',','_')  as terminal_dispatch_code
         , shipment_arrival_time
         , dt
    from jms_dm.dm_cn_departure_transfer_timely_route_dt
    where dt = '{{ execution_date | cst_ds }}'
),
     waybill_incre as (
         select waybill_no
         from jms_dwd.dwd_yl_oms_oms_waybill_incre_dt
         where dt >= date_add('{{ execution_date | cst_ds }}', -15)
           and dt <= '{{ execution_date | cst_ds }}'
           and express_type_name = '兔优达'
     )
select if(t1.billcode = '', null, t1.billcode)                                   as tbillcode
     , if(t1.start_code = '', null, t1.start_code)                               as tstart_code
     , if(t1.start_name = '', null, t1.start_name)                               as tstart_name
     , if(t1.order_source_code = '', null, t1.order_source_code)                 as torder_source_code
     , if(t1.order_source_name = '', null, t1.order_source_name)                 as torder_source_name
     , if(t1.start_agent_code = '', null, t1.start_agent_code)                   as tstart_agent_code
     , if(t1.start_agent_name = '', null, t1.start_agent_name)                   as tstart_agent_name
     , if(t1.center_type = '', null, t1.center_type)                             as tcenter_type
     , if(t1.nextstationcode = '', null, t1.nextstationcode)                     as tnextstationcode
     , if(t1.nextstation = '', null, t1.nextstation)                             as tnextstation
     , if(t1.send_scantime = '', null, t1.send_scantime)                         as tsend_scantime
     , if(t1.send_shipment_no = '', null, t1.send_shipment_no)                   as tsend_shipment_no
     , if(t1.planned_departure_time = '', null, t1.planned_departure_time)       as tplanned_departure_time
     , if(t1.actual_departure_time = '', null, t1.actual_departure_time)         as tactual_departure_time
     , if(t1.next_planned_arrival_time = '', null, t1.next_planned_arrival_time) as tnext_planned_arrival_time
     , if(t1.packagecode = '', null, t1.packagecode)                             as tpackagecode
     , if(t1.goods_type_code = '', null, t1.goods_type_code)                     as tgoods_type_code
     , if(t1.goods_type_name = '', null, t1.goods_type_name)                     as tgoods_type_name
     , if(t1.arrival_shipment_no = '', null, t1.arrival_shipment_no)             as tarrival_shipment_no
     , if(t1.arrival_time = '', null, t1.arrival_time)                           as tarrival_time
     , if(t1.all_line_arrival_time = '', null, t1.all_line_arrival_time)         as tall_line_arrival_time
     , if(t1.last_network_code = '', null, t1.last_network_code)                 as tlast_network_code
     , if(t1.last_network_name = '', null, t1.last_network_name)                 as tlast_network_name
     , if(t1.transfer_hours = '', null, t1.transfer_hours)                       as ttransfer_hours
     , if(t1.route_send_time = '', null, t1.route_send_time)                     as troute_send_time
     , if(t1.all_line_route_send_time = '', null, t1.all_line_route_send_time)   as tall_line_route_send_time
     , if(t1.is_timely = '', null, t1.is_timely)                                 as tis_timely
     , if(t1.untimely_cause = '', null, t1.untimely_cause)                       as tuntimely_cause
     , if(t1.all_line_is_timely = '', null, t1.all_line_is_timely)               as tall_line_is_timely
     , if(t1.all_line_untimely_cause = '', null, t1.all_line_untimely_cause)     as tall_line_untimely_cause
     , if(t1.route_lack = '', null, t1.route_lack)                               as troute_lack
     , if(t1.route_error = '', null, t1.route_error)                             as troute_error
     , if(t1.end_code = '', null, t1.end_code)                                   as tend_code
     , if(t1.end_name = '', null, t1.end_name)                                   as tend_name
     , if(t1.r_nxt_arrival_time = '', null, t1.r_nxt_arrival_time)               as tr_nxt_arrival_time
     , if(t1.all_line_r_nxt_arrival_time = '', null,
          t1.all_line_r_nxt_arrival_time)                                        as tall_line_r_nxt_arrival_time
     , if(t1.r_lastes_arrival_time = '', null, t1.r_lastes_arrival_time)         as tr_lastes_arrival_time
     , if(t1.all_line_r_lastes_arrival_time = '', null,
          t1.all_line_r_lastes_arrival_time)                                     as tall_line_r_lastes_arrival_time
     , if(t1.stop_type = '', null, t1.stop_type)                                 as tstop_type
     , if(t1.pack_send_time = '', null, t1.pack_send_time)                       as tpack_send_time
     , if(t1.mage_region_code = '', null, t1.mage_region_code)                   as tmage_region_code
     , if(t1.mage_region_name = '', null, t1.mage_region_name)                   as tmage_region_name
     , if(t1.business_attribute = '', null, t1.business_attribute)               as tbusiness_attribute
     , if(t1.r_next_lastes_arrival_time = '', null,
          t1.r_next_lastes_arrival_time)                                         as tr_next_lastes_arrival_time
     , if(t1.all_r_next_lastes_arrival_time = '', null,
          t1.all_r_next_lastes_arrival_time)                                     as tall_r_next_lastes_arrival_time
     , if(t1.arrival_actual_arrival_time = '', null,
          t1.arrival_actual_arrival_time)                                        as tarrival_actual_arrival_time
     , if(t1.arrival_planned_arrival_time = '', null,
          t1.arrival_planned_arrival_time)                                       as tarrival_planned_arrival_time
     , if(t1.reserved_seconds = '', null, t1.reserved_seconds)                   as treserved_seconds
     , if(t1.sign_agent_code = '', null, t1.sign_agent_code)                     as tsign_agent_code
     , if(t1.sign_agent_name = '', null, t1.sign_agent_name)                     as tsign_agent_name
     , if(t1.end_center_code = '', null, t1.end_center_code)                     as tend_center_code
     , if(t1.end_center_name = '', null, t1.end_center_name)                     as tend_center_name
     , if(t1.center_arrvial_time = '', null, t1.center_arrvial_time)             as tcenter_arrvial_time
     , if(t1.terminal_dispatch_code = '', null, t1.terminal_dispatch_code)       as tterminal_dispatch_code
     , if(t1.shipment_arrival_time = '', null, t1.shipment_arrival_time)         as tshipment_arrival_time
from departure_transfer t1
         join waybill_incre t2
              on t1.billcode = t2.waybill_no
;
 " >>/tmp/${name}_${dt}.txt
}

# 替换分隔符，并输出csv
cat /tmp/${name}_${dt}.txt | tr -s '\t' ',' > /tmp/${name}_${dt}.csv
# 压缩并删除 csv 文件
zip -jrm /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv
# 上传文件
/usr/local/python3/bin/python3  /usr/local/airflow/dags/utils/operators/upload_oss.py ${name}/${name}_${dt}.zip /tmp/${name}_${dt}.zip
# zip 文件将在进程退出后被自动清理

#声明文件大小
data_size=$(ls -lh /tmp/${name}_${dt}.zip | awk '{print $5}')
#打印文件信息
Message="${name}_${dt}.zip 上传成功 大小: $data_size"
#配置钉钉告警
curl 'https://oapi.dingtalk.com/robot/send?access_token=816a8f35fd84432ffa6bff9d7714abc322628bb4a8ded86908550d2335aba613' \
   -H 'Content-Type: application/json' \
   -d "
  {\"msgtype\": \"text\",
    \"text\": {
        \"content\": \"监控内容:$Message\"
     }
  }" || true
rm -f  /tmp/${name}_${dt}.txt